package com.shujia.flink.state;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Hashtable;

public class Demo02CheckPoint {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 开启CheckPoint：每 10000ms 开始一次 checkpoint
        env.enableCheckpointing(10000);
        // 高级选项：
        // 设置模式为精确一次 (这是默认值)
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 两次checkpoints 之间的最小延迟
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        // Checkpoint 必须在一分钟内完成，否则就会被抛弃
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        // 允许两个连续的 checkpoint 错误
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(2);
        // 同一时间只允许一个 checkpoint 进行
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // 使用 externalized checkpoints，这样 checkpoint 在作业取消后仍就会被保留
        env.getCheckpointConfig().setExternalizedCheckpointCleanup(
                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //将状态保存到hdfs中
        env.getCheckpointConfig().setCheckpointStorage("hdfs://master:9000/flink/checkpoint");



        DataStreamSource<String> socketDS = env.socketTextStream("master", 8888);


//        socketDS.map(w -> Tuple2.of(w, 1), Types.TUPLE(Types.STRING, Types.INT)).print();

        // 自定义处理逻辑：可以实现map、flatMap、filter等等操作
        // 先切分，进行扁平化处理，判断进来的单词是否等于vip，如果是则将其变成vip 10，其他的转换成 word 1，并且如果单词=b，则不输出
        SingleOutputStreamOperator<Tuple2<String, Integer>> kvDS = socketDS.process(new ProcessFunction<String, Tuple2<String, Integer>>() {
            // 每一条数据都会执行一次
            @Override
            /*
            line : 接收到的每一条数据
            context ：上下文环境，主要用于获取时间戳
            out：用于将数据发送到下游
             */
            public void processElement(String line, ProcessFunction<String, Tuple2<String, Integer>>.Context context, Collector<Tuple2<String, Integer>> out) throws Exception {
                for (String word : line.split(",")) {
                    if (!"b".equals(word)) {
                        int value = 0;
                        switch (word) {
                            case "vip":
                                value = 10;
                                break;
                            case "svip":
                                value = 100;
                                break;
                            default:
                                value = 1;
                                break;
                        }
                        // 通过out将数据以KV格式发送到下游
                        out.collect(Tuple2.of(word, value));
                    }
                }

            }
        });

        kvDS.keyBy(t2 -> t2.f0)
                .process(new KeyedProcessFunction<String, Tuple2<String, Integer>, String>() {
                    // 在每个并行度对应的线程中会共用该值
//                    Integer sumValue = 0;
                    // 给每个Key单独维护一个状态值，借助HashMap来实现
                    // 程序重启后状态会丢失
//                    Hashtable<String, Integer> kvStateMap;
                    // 使用Flink提供的状态，会自动的给每一个Key维护一个状态
                    ValueState<Integer> cntState;

                    @Override
                    // Job启动时每个并行度对应的线程中运行的Task会执行一次
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        System.out.println("线程：" + Thread.currentThread().getId() + "进行了初始化");
                        // 通过Flink的运行环境初始化（第一次运行）或者是加载（恢复）状态
                        cntState = getRuntimeContext().getState(new ValueStateDescriptor<Integer>("cntState", Types.INT));
                    }

                    @Override
                    // 每一条数据会执行一次
                    public void processElement(Tuple2<String, Integer> kv, KeyedProcessFunction<String, Tuple2<String, Integer>, String>.Context context, Collector<String> out) throws Exception {
                        // 统计每个单词的value之和
                        int newValue = 0;
                        if (cntState.value() == null){
                            newValue = kv.f1;
                        }else {
                            newValue = cntState.value() + kv.f1; // 取之前的状态 + 当前这一条数据的value
                        }

                        cntState.update(newValue); // 更新状态
                        out.collect(kv.f0 + "," + newValue);
                    }
                }).print();

        /*
         * 任务提交的步骤：
         * 1、打包上传
         * 2、提交任务：
         *      第一次提交任务：
         *          WEB页面：
         *              主类名：com.shujia.flink.state.Demo02CheckPoint
         *              并行度：1
         *          命令行：
         *              # application_1744427567611_0007：yarn-session启动之后在Yarn的ApplicationID，可以从8088页面获取
         *              flink run -t yarn-session -p 1 -Dyarn.application.id=application_1744427567611_0007 -c com.shujia.flink.state.Demo02CheckPoint flink-1.0-SNAPSHOT.jar
         *      失败之后的恢复：
         *          WEB界面：
         *              主类名：com.shujia.flink.state.Demo02CheckPoint
         *              并行度：1
         *              # c4433e1bf92214a3d8e971d937e48980：任务ID
         *              # chk-7：表示已经完成了7次CK
         *              # 每个人的任务ID以及完成次数都不相同，自己根据情况调整
         *              savepoint path： hdfs://master:9000/flink/checkpoint/c4433e1bf92214a3d8e971d937e48980/chk-7
         *          命令行：
         *              # 通过-s参数指定恢复的路径
         *              flink run -t yarn-session -s hdfs://master:9000/flink/checkpoint/209b0bbb203a6564ed2dfb02df89fd37/chk-7 -p 1 -Dyarn.application.id=application_1744427567611_0007 -c com.shujia.flink.state.Demo02CheckPoint flink-1.0-SNAPSHOT.jar
         */


        env.execute();

    }
}
